package org.example.window;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/19 15:53
 */
public class WindowReduceDemo01 {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = environment.socketTextStream("127.0.0.1", 7777);

        KeyedStream<WaterSensor, String> keyedStream = streamSource.map(new WaterSensorFunction())
                .keyBy(WaterSensor::getId);

        keyedStream
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<WaterSensor>() {
                    @Override
                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                        System.out.println("value1--->" + value1 + "value2--->" + value2);
                        return new WaterSensor(value1.id, value1.vt, value1.vc + value2.vc);
                    }
                })
                .print()
        ;

        environment.execute();
    }

}
